
#include <iostream>
#include <sstream>
#include <iomanip>
#include <string>
#include <vector>

#include <boost/filesystem.hpp>
#include <boost/program_options.hpp>
namespace po = boost::program_options;


#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;

#include "bocat/common/md5.hpp"
#include "bocat/common/logging.hpp"

#include "boque/tranzmq/client.hpp"

using namespace std;

#define	DEFAULT_TIMES	32

#define	N_ROUND	100000

static bocat::common::Logger*   logger_p = NULL;

#define INFO(ARGS...)   logger_p->info(__LINE__,__FILE__,ARGS)
#define ERROR(ARGS...)  logger_p->error(__LINE__,__FILE__,ARGS)

float
getTimeDuration( ptime time_begin, ptime time_end )
{
	time_period	tp( time_begin, time_end );
	time_duration	td = tp.length();
	long	time_last_ms = td.total_milliseconds();
	float	time_last_s = time_last_ms / 1000.0f;

	return time_last_s;
}



void	test_001( const std::string& addr_to, const std::string& buf_dir, int times, int data_size )
{
	boque::tranzmq::Client	clt( buf_dir.c_str() );

	char*	buf = new char[ data_size ];
	memset( buf, 0, data_size );
	memset( buf, 'A', data_size - 1 );

	// test SEND
	{
		INFO( "testing SEND ..." );
		ptime	time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;
		u_int32_t	count_round = 0;

		long	bytes_total = 0;
		for (int i=0; i<times; i++) {
			ostringstream	os_data;     os_data << "DATA--" << setw(8) << setfill('0') << i << buf;
			string	data = os_data.str();

			clt.send( addr_to, data );

			bytes_total += data.size();

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
                INFO("  R-SEND %u*%u records last %.3f seconds, tps=%.2f", ++count_round, N_ROUND, time_last_s, tps);

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ptime	time_stop = microsec_clock::local_time();
		float	time_last_s = getTimeDuration( time_start, time_stop );
		float	tps = times / time_last_s;
		float	bw = bytes_total / time_last_s / 1048576.0;
        INFO("    SEND %u records last %.3f seconds, tps=%.2f, bw=%.2f MB/s", times, time_last_s, tps, bw);
	}


}


int	main(int argc, char* argv[])
{
	int 	times = 0;
	int 	data_size = 0;
	bool	verbose = false;
	std::string	buf_dir;
	std::string	addr_to;

    try {

        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("verbose,v", "output verbose message")
			("times", po::value<int>(&times)->default_value(DEFAULT_TIMES), "number of test to execute")
			("data-size", po::value<int>(&data_size)->default_value( 32 ), "size of data chunk")
            ("buf-dir", po::value<std::string>(&buf_dir)->default_value("/tmp/boque/tranzmq"), "dir to buffer data")
            ("to", po::value<std::string>(&addr_to)->default_value("tcp://127.0.0.1:5555"), "sink endpoint")
        ;   

        po::variables_map vm;    
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);    

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }   

		if (vm.count("verbose")) {
			verbose = true;
		}
    }   
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }   
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
		return 2;
    }   

    logger_p = bocat::common::getLogger();

	
	test_001( addr_to, buf_dir, times, data_size );


	return	0;
}

